<?php

namespace Tyrone\Lib;


class ConsumeWorker extends BaseWorker
{

    private $flag;
    private $tasks;
    private $queue;
    private $class;
    private $CenterMysqlPool;
    private $stream;
    private $group;
    private $type = 'consume';
    public function init(&$server, $worker_id, $conf)
    {
        $this->loadBaseFile($conf);
        $this->tasks = require DIR.'/data/'.$this->type.'.php';
        $this->worker_id = $worker_id;

        $this->RedisPool = $this->creatRedisPool($conf);
        foreach ($this->tasks as $name => $task){
            $worker_ids = $task['worker_id'];
            $this->MysqlPool = $this->creatMysqlPool($conf,'log_center', count($worker_ids));
            $this->CenterMysqlPool = $this->creatMysqlPool($conf,'center', count($worker_ids));
            if(!in_array($worker_id,$worker_ids)){
                continue;
            }
            $this->class = $task['class'];
            $this->queue = $task['redis']['queue'] ?? '';
            $this->stream = $task['redis']['stream'] ?? '';
            $this->group = $task['redis']['group'] ?? '';
            $this->flag = $task['redis']['limit'] ?? 1;
        }

        if($this->type=='stream'){
            $this->redisStream($worker_id,$conf);
        }else{
            $server->ticks[$worker_id][] = $this->tick();
        }

    }

    // 出队列
    public function getQueue()
    {
        $redis = $this->RedisPool->get();
        //var_dump($this->queue);
        $data = $redis->rpop($this->queue);
        $this->RedisPool->put($redis);
        //print_r($data);
        return $data;
    }

    // 入队列
    public function lpushQueue($queue, $data)
    {
        $redis = $this->RedisPool->get();
        $data = $redis->lpush($queue,$data);
        $this->RedisPool->put($redis);
        return $data;
    }


    public function handleTask($task){
        if (empty($task))
        {
            return;
        }
        // 主动关闭进程
        if ('exit' == $task)
        {
            exit(0);
        }else
        {
            $mysql = $center_mysql = null;
            try {
                $mysql = $this->MysqlPool->get();
                $center_mysql = $this->CenterMysqlPool->get();

                defer(function () use ($mysql,$center_mysql) {
                    if ($mysql && $this->MysqlPool) {
                        $this->MysqlPool->put($mysql);
                    }
                    if ($center_mysql && $this->CenterMysqlPool) {
                        $this->CenterMysqlPool->put($mysql);
                    }
                });
                $class_name = $this->class;
                $class = new $class_name($this->RedisPool);
                //echo "开始执行{$mysqlKey}:{$database}:{$class_name}:{$method}的\n";
                $class->run($task, $mysql,$center_mysql);
            } catch (\Throwable $throwable) {
                echo "Task error: " . $throwable->getMessage() . "\n";
                echo "Trace: " . $throwable->getTraceAsString() . "\n";
                echo "Trace: " . $throwable->getLine() . "\n";
                echo "task: " . $task . "\n";
            } finally {

            }
        }
    }

    public function tick()
    {
        return \Swoole\Timer::tick(RATE ?? 1000, function () {
            $i = $this->flag;
            while ($i)
            {
                try{
                    // 从redis队列中出栈的内容
                    $task = $this->getQueue();
                }catch (\Throwable $e){
                    $task = '';
                    var_dump($e->getTrace());
                }
                if($task){
                    \Swoole\Coroutine::create(function () use ($task) {
                        $this->handleTask($task ?? '');
                    });
                }
                --$i;
            }
        });
    }

    public function redisStream($workerId,$conf){
        $redis = $this->creatRedis($conf);
        print_r($redis);
        $stream = $this->stream;
        $group = $this->group;
        $consumer = GAME_NAME.'_'.APP_NAME.'_group_'.GROUP.'_consumer_worker'.$workerId;
        // 创建组（如果不存在）
        $redis->xGroupCreate($stream, $group, '0', true);
        var_dump($stream);
        var_dump($group);
        var_dump($consumer);
        while (true) {
            try {
                $streamList = $redis->xReadGroup($group,                    // 第1个参数：group_name
                    $consumer,                 // 第2个参数：consumer_name
                    [$stream => '>'],          // 第3个参数：streams 数组
                    [                          // 第4个参数：options 数组（包含 count 和 block）
                        'count' => $this->flag,
                        'block' => RATE ?? 1000
                    ]);

                if ($streamList) {
                    echo date('Y-m-d H:i:s').PHP_EOL;
                        foreach ($streamList as  $streamData) {
                            list($cstream,$messages) = $streamData;
                            foreach ($messages as $message){
                                \Swoole\Coroutine::create(function () use ($message,$stream,$group,$consumer) {
                                        $redis = null;
                                        list($id,$data) = $message;
                                        try {
                                            $redis = $this->RedisPool->get();

                                            $class_name = $this->class;
                                            $class = new $class_name($this->RedisPool);
                                            $params = [$redis,$message];
                                            $result = $class->run(...$params);
                                            $result && $redis->xAck($stream, $group, [$id]);
                                        } catch (\Exception $e) {
                                            echo "Task error: " . $e->getMessage() . "\n";
                                            echo "Trace: " . $e->getTraceAsString() . "\n";
                                            echo "Trace: " . $e->getLine() . "\n";
                                            echo "task: " . json_encode($message) . "\n";
                                        } finally {
                                            if ($redis && $this->RedisPool) {
                                                $this->RedisPool->put($redis);
                                            }
                                        }
                                    });
                            }

                        }
                }
            } catch (\Exception $e) {
                error_log("Read error: " . $e->getMessage());
                \Swoole\Coroutine::sleep(1);
            }
        }
    }


}